package core

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"net"
	neturl "net/url"
	"strings"
	"time"

	"yunshicloud.com/gproxy/core/amf"

	"github.com/rs/zerolog/log"
)

var (
	respResult     = "_result"
	respError      = "_error"
	onStatus       = "onStatus"
	publishStart   = "NetStream.Publish.Start"
	playStart      = "NetStream.Play.Start"
	connectSuccess = "NetConnection.Connect.Success"
	onBWDone       = "onBWDone"
)

var (
	ErrFail = fmt.Errorf("respone err")
)

type ConnClient struct {
	done       bool
	transID    int
	url        string
	tcurl      string
	app        string
	title      string
	query      string
	curcmdName string
	streamid   uint32
	conn       *Conn
	encoder    *amf.Encoder
	decoder    *amf.Decoder
	bytesw     *bytes.Buffer
}

func NewConnClient() *ConnClient {
	return &ConnClient{
		transID: 1,
		bytesw:  bytes.NewBuffer(nil),
		encoder: &amf.Encoder{},
		decoder: &amf.Decoder{},
	}
}

func (connClient *ConnClient) DecodeBatch(r io.Reader, ver amf.Version) (ret []interface{}, err error) {
	vs, err := connClient.decoder.DecodeBatch(r, ver)
	return vs, err
}

func (connClient *ConnClient) readRespMsg() error {
	var err error
	var rc ChunkStream
	defer connClient.conn.SetDeadline(time.Time{})
	defer time.Sleep(time.Duration(20) * time.Millisecond)

	for {
		//设置最大的timeout时间为2秒钟
		connClient.conn.SetDeadline(time.Now().Add(2 * time.Second))
		err = connClient.conn.Read(&rc)
		if err != nil {
			log.Error().Int32("chunk_size", int32(connClient.conn.remoteChunkSize)).Int32("windowAckSize", int32(connClient.conn.remoteWindowAckSize)).Err(err).Msg("readRespMsg错误")
			return err
		}
		if err != nil && err != io.EOF {
			return err
		}
		switch rc.TypeID {
		case 20, 17:
			r := bytes.NewReader(rc.Data)
			vs, _ := connClient.decoder.DecodeBatch(r, amf.AMF0)

			log.Debug().Interface("vs", vs).Msg("readRespMsg")
			for k, v := range vs {
				switch v.(type) {
				case string:
					switch connClient.curcmdName {
					case cmdConnect, cmdCreateStream:
						if v.(string) != respResult {
							return fmt.Errorf(v.(string))
						}

					case cmdPublish:
						if v.(string) != onStatus {
							return ErrFail
						}
					}
				case float64:
					switch connClient.curcmdName {
					case cmdConnect, cmdCreateStream:
						id := int(v.(float64))

						if k == 1 {
							if id != connClient.transID {
								return ErrFail
							}
						} else if k == 3 {
							connClient.streamid = uint32(id)
						}
					case cmdPublish:
						if int(v.(float64)) != 0 {
							return ErrFail
						}
					}
				case amf.Object:
					objmap := v.(amf.Object)
					switch connClient.curcmdName {
					case cmdConnect:
						code, ok := objmap["code"]
						if ok && code.(string) != connectSuccess {
							return ErrFail
						}
					case cmdPublish:
						code, ok := objmap["code"]
						if ok && code.(string) != publishStart {
							return ErrFail
						}
					}
				}
			}
			return nil
		case 1:
			//if ID=1,get chunksize
			cs := amf.BytesToInt32(rc.Data)
			log.Debug().Int32("checksize", cs)
			connClient.conn.remoteChunkSize = uint32(cs)
		case 5:
			//if
			log.Debug().Msg("5")
		case 6:
			log.Debug().Msg("6")
		}
	}
}

func (connClient *ConnClient) writeMsg(args ...interface{}) error {
	connClient.bytesw.Reset()
	for _, v := range args {
		_, err := connClient.encoder.Encode(connClient.bytesw, v, amf.AMF0)
		if err != nil {
			return err
		}
	}
	msg := connClient.bytesw.Bytes()
	c := ChunkStream{
		Format:    0,
		CSID:      3,
		Timestamp: 0,
		TypeID:    20,
		StreamID:  connClient.streamid,
		Length:    uint32(len(msg)),
		Data:      msg,
	}
	connClient.conn.Write(&c)
	return connClient.conn.Flush()
}

func (connClient *ConnClient) writeConnectMsg() error {
	event := make(amf.Object)
	event["app"] = connClient.app
	event["type"] = "nonprivate"
	event["flashVer"] = "FMS.3.1"
	event["tcUrl"] = connClient.tcurl
	connClient.curcmdName = cmdConnect

	log.Debug().Int("connClient.transID", connClient.transID)
	err := connClient.writeMsg(cmdConnect, connClient.transID, event)
	if err != nil {
		return err
	}
	return connClient.readRespMsg()
}

func (connClient *ConnClient) writeCreateStreamMsg() error {
	connClient.transID++
	connClient.curcmdName = cmdCreateStream

	time.Sleep(time.Duration(20) * time.Millisecond)
	log.Debug().Int("connClient.transID", connClient.transID).Msg("writeCreateStreamMsg")
	err := connClient.writeMsg(cmdCreateStream, connClient.transID, nil)
	if err != nil {
		return err
	}

	for {
		err := connClient.readRespMsg()
		if err == nil {
			return err
		}

		if err == ErrFail {
			log.Error().Err(err)
			return err
		}
	}

}

func (connClient *ConnClient) writePublishMsg() error {
	connClient.transID++
	connClient.curcmdName = cmdPublish
	err := connClient.writeMsg(cmdPublish, connClient.transID, nil, connClient.title, "live")
	if err != nil {
		return err
	}
	return connClient.readRespMsg()
}

func (connClient *ConnClient) writePlayMsg() error {
	connClient.transID++
	connClient.curcmdName = cmdPlay

	log.Debug().Int("connClient.transID", connClient.transID).Str("cmdPlay", cmdPlay).Str("connClient.title", connClient.title).Msg("writePlayMsg")

	time.Sleep(time.Duration(20) * time.Millisecond)
	err := connClient.writeMsg(cmdPlay, 0, nil, connClient.title)
	if err != nil {
		log.Error().Err(err).Msg("writePlayMsg")
		return err
	}
	time.Sleep(time.Duration(20) * time.Millisecond)
	log.Debug().Int("connClient.transID", connClient.transID).Str("cmdPlay", cmdPlay).Str("connClient.title", connClient.title).Msg("开始读取服务器的回传信息")
	return connClient.readRespMsg()

}

func (connClient *ConnClient) HandshakeClient(url string) error {
	u, err := neturl.Parse(url)
	if err != nil {
		return err
	}
	connClient.url = url
	path := strings.TrimLeft(u.Path, "/")
	ps := strings.SplitN(path, "/", 2)
	if len(ps) != 2 {
		return fmt.Errorf("u path err: %s", path)
	}
	connClient.app = ps[0]
	connClient.title = ps[1]
	connClient.query = u.RawQuery
	connClient.tcurl = "rtmp://" + u.Host + "/" + connClient.app
	port := ":1935"
	host := u.Host
	localIP := ":0"
	var remoteIP string
	if strings.Contains(host, ":") {
		host, port, err = net.SplitHostPort(host)
		if err != nil {
			return err
		}
		port = ":" + port
	}
	ips, err := net.LookupIP(host)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	remoteIP = ips[rand.Intn(len(ips))].String()
	if !strings.Contains(remoteIP, ":") {
		remoteIP += port
	}

	local, err := net.ResolveTCPAddr("tcp", localIP)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	log.Debug().Str("remoteIP: ", remoteIP)
	remote, err := net.ResolveTCPAddr("tcp", remoteIP)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	conn, err := net.DialTCP("tcp", local, remote)
	if err != nil {
		log.Error().Err(err)
		return err
	}

	log.Debug().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Msg("connection")
	connClient.conn = NewConn(conn, 4*1024)

	log.Debug().Msg("HandshakeClient....")
	err = connClient.conn.HandshakeClient()
	if err != nil {
		return err
	}
	log.Debug().Msg("writeConnectMsg....")
	err = connClient.writeConnectMsg()
	if err != nil {
		return err
	}

	return nil
}

func (connClient *ConnClient) Play() error {

	log.Debug().Msg("writeCreateStreamMsg....")
	err := connClient.writeCreateStreamMsg()
	if err != nil {
		log.Error().Err(err)
		return err
	}

	log.Debug().Msg("method control:PLAY")
	err = connClient.writePlayMsg()
	if err != nil {
		return err
	}
	return nil
}

func (connClient *ConnClient) Start(url string, method string) error {
	//首先处理该问的流地址；
	u, err := neturl.Parse(url)
	if err != nil {
		return err
	}
	connClient.url = url
	path := strings.TrimLeft(u.Path, "/")
	ps := strings.SplitN(path, "/", 2)
	if len(ps) != 2 {
		return fmt.Errorf("u path err: %s", path)
	}
	connClient.app = ps[0]
	connClient.title = ps[1]
	connClient.query = u.RawQuery
	connClient.tcurl = "rtmp://" + u.Host + "/" + connClient.app
	port := ":1935"
	host := u.Host
	localIP := ":0"
	var remoteIP string
	if strings.Contains(host, ":") {
		host, port, err = net.SplitHostPort(host)
		if err != nil {
			return err
		}
		port = ":" + port
	}
	ips, err := net.LookupIP(host)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	remoteIP = ips[rand.Intn(len(ips))].String()
	if !strings.Contains(remoteIP, ":") {
		remoteIP += port
	}

	local, err := net.ResolveTCPAddr("tcp", localIP)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	log.Debug().Str("remoteIP: ", remoteIP)
	remote, err := net.ResolveTCPAddr("tcp", remoteIP)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	// time.Sleep(time.Duration(50) * time.Millisecond)
	log.Debug().Str("from:", "local").Str("to:", remote.String()).Msg("gproxy与流媒体服务器建立新连接")
	conn, err := net.DialTCP("tcp", local, remote)
	if err != nil {
		log.Error().Err(err)
		return err
	}
	// time.Sleep(time.Duration(50) * time.Millisecond)
	connClient.conn = NewConn(conn, 4*1024)

	log.Debug().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Msg("gproxy开始与流媒体服务器进行握手")
	// time.Sleep(time.Duration(10) * time.Millisecond)
	err = connClient.conn.HandshakeClient()
	if err != nil {
		log.Error().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Err(err).Msg("gproxy与流媒体服务器握手失败")
		return err
	}

	log.Debug().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Msg("gproxy与流媒体服务器握手成功")
	log.Debug().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Msg("gproxy与流媒体服务器send:connect")

	err = connClient.writeConnectMsg()
	if err != nil {
		log.Error().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Err(err).Msg("send:connect")
		return err
	}

	log.Debug().Str("local:", conn.LocalAddr().String()).Str("remote:", conn.RemoteAddr().String()).Msg("gproxy创建流信息")
	err = connClient.writeCreateStreamMsg()
	if err != nil {
		log.Error().Err(err)
		return err
	}

	log.Debug().Msg("method control:PUBLISH,PLAY")
	if method == "publish" {
		return errors.New("不支持publish")
	} else if method == "play" {
		err = connClient.writePlayMsg()
		if err != nil {
			return err
		}
	}

	return nil
}

func (connClient *ConnClient) Write(c ChunkStream) error {
	if c.TypeID == 18 ||
		c.TypeID == 0xf {
		var err error
		c.Data, err = amf.MetaDataReform(c.Data, amf.ADD)
		if err != nil {
			return err
		}
		c.Length = uint32(len(c.Data))
	}
	return connClient.conn.Write(&c)
}

func (connClient *ConnClient) Flush() error {
	return connClient.conn.Flush()
}

func (connClient *ConnClient) Read(c *ChunkStream) (err error) {
	return connClient.conn.Read(c)
}

func (connClient *ConnClient) GetInfo() (app string, name string, url string) {
	app = connClient.app
	name = connClient.title
	url = connClient.url
	return
}

func (connClient *ConnClient) GetStreamId() uint32 {
	return connClient.streamid
}

func (connClient *ConnClient) Close(err error) {
	connClient.conn.Close()
}
